package com.rec.smile.rpc.registry.redis;

import java.net.URI;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadLocalRandom;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPubSub;

import com.rec.smile.rpc.api.util.Constant;
import com.rec.smile.rpc.registry.RootRegistryApi;

public class RedisDiscovery extends RootRegistryApi {

	private JedisPool jedisPool;
	
	private List<String> messages = new CopyOnWriteArrayList<String>();
	
	public RedisDiscovery(String redisServerAddressUri){
		jedisPool = new JedisPool(URI.create(redisServerAddressUri));
		
		try(Jedis jedis = jedisPool.getResource()){
			jedis.subscribe(new JedisPubSub() {

				@Override
				public void onMessage(String channel, String message) {
//					super.onMessage(channel, message);
					if(Constant.ZK_DATA_PATH.equals(channel)){
						messages.add(message);
						while(messages.size() > 1000){
							messages.remove(0);
						}
						
					}
				}

				@Override
				public void onSubscribe(String channel, int subscribedChannels) {
					super.onSubscribe(channel, subscribedChannels);
				}
				
				
			}, Constant.REDIS_DATA_PATH);
		}catch (Exception e) {
		}
	}
	
	@Override
	public String discover() {
		if(messages.isEmpty()){
			return null;
		}
		
		return messages.get(ThreadLocalRandom.current().nextInt(messages.size()));
	}

}
